/*
 * scc.c, SmartCam skeleton system remote operations wrapper
 *
 */

#define _XOPEN_SOURCE 500

#define __LIBSCC
#define LIBMAIN

#include <string.h>
#include <errno.h>
#include <log.h>
#include <netinet/tcp.h>
                     
#include "sc.h"
#include "scc.h"
#include "scu.h"
#include "management.h"

#define NODELAY

log_default_streams_t sccl;
table_t scc_procs;
scc_proc_t scc_master;
struct sockaddr scc_address;
int scc_addrlen;
scos_mutex_t scc_mutex;

int addrcmp(struct sockaddr *addr1, int addrlen1, struct sockaddr *addr2, int addrlen2);
void sccWriteHeader(scc_hdr_t *hdr);

void sccInit(struct sockaddr *myaddr, int addrlen)
{
  int i=0, id=-1;
  int single=1;
  scc_proc_t *p;
  scos_thread_t thread;
  sc_proc_t *scp;

  logAddGroup();
  sccl = logAddDefaultStreams(NULL);
  logEdit(sccl.info, LOG_GROUP_SET_LEVEL);

  lprintf(STREAM(sccl.notice), "Initializing communications driver");

  memcpy(&scc_address, myaddr, sizeof(scc_address));
  scc_addrlen = addrlen;
  abortOnNegative(STREAM(sccl.err), tblcreate(&scc_procs, sizeof(scc_proc_t)));

  scos_mutex_init(&scc_mutex);

  scc_master.id = SCC_MASTER_PROCESSOR;
  scc_master.name = "MCP";
  scc_master.control_fd = scc_master.incoming_data_fd = scc_master.outgoing_data_fd = -1;

  while (tblgetnextelement(&sc_procs, &id, &scp))
  {
    abortOnNegative(STREAM(sccl.err), tbladdnewi(&scc_procs, &p, id));
    p->id = i;
    abortOnNULL(STREAM(sccl.err), p->name = malloc(strlen(scp->name)+1));
    strcpy(p->name, scp->name);
    memcpy(&p->address, &scp->address, sizeof(p->address));
    p->addrlen = scp->addrlen;
    p->control_fd = p->incoming_data_fd = p->outgoing_data_fd = -1;
    i++;

    if (addrcmp(&scc_address, scc_addrlen, &p->address, p->addrlen))
      single = 0;
  }

  if (!single && !settings.simulate)
  {
    /* Start incoming data connection thread */
    scos_thread_create(&thread, sccIncomingConnectionThread, NULL);
  }

  sccTransportInit();
}

void sccSendMessage(scc_hdr_t *hdr)
{
  scc_proc_t *d;

  if (hdr->destination == SCC_MASTER_PROCESSOR) d = &scc_master;
  else abortOnNULL(STREAM(sccl.err), tblget(&scc_procs, hdr->destination, &d));

  lprintf(STREAM(sccl.crawl), "Sending message type %d to %s",
	  hdr->type, d->name);

  /* Local communications */
  switch (hdr->type)
  {
    case SCC_MSGTYPE_BUFFERDATA:
    case SCC_MSGTYPE_BUFFERDRY:
    case SCC_MSGTYPE_BUFFERACK:
      if (!addrcmp(&scc_address, scc_addrlen, &d->address, d->addrlen) || settings.simulate)
	d->datacallback(hdr);
      else
      {
        /* http://www.almaden.ibm.com/cs/people/marksmith/sendmsg.html */
	scos_mutex_lock(&scc_mutex);
	if (d->outgoing_data_fd < 0) sccOpenDataConnection(d);

	abortOnError(STREAM(sccl.err),
		     scos_send(d->outgoing_data_fd, (char*)hdr, sizeof(scc_hdr_t)+hdr->size, 0)
		     != sizeof(scc_hdr_t)+hdr->size);

	scos_mutex_unlock(&scc_mutex);
      }
      break;
    default:
      if (!addrcmp(&scc_address, scc_addrlen, &d->address, d->addrlen) || settings.simulate)
	d->controlcallback(hdr);
      else
      {
	scos_mutex_lock(&scc_mutex);
	abortOnError(STREAM(sccl.err),
		     scos_send(d->control_fd, (char*)hdr, sizeof(scc_hdr_t)+hdr->size, 0)
		     != sizeof(scc_hdr_t)+hdr->size);
	scos_mutex_unlock(&scc_mutex);
      }
  }

  lprintf(STREAM(sccl.crawl), "Message type %d to %s sent", hdr->type, d->name);
}

void sccMasterSetup(scc_controlcallback_t controlcallback, scc_datacallback_t datacallback)
{
  int id=-1;
  scc_proc_t *p;

  lprintf(STREAM(sccl.notice), "Setting up master processor");

  memcpy(&scc_master.address, &scc_address, scc_addrlen);
  scc_master.addrlen = scc_addrlen;
  scc_master.controlcallback = controlcallback;
  scc_master.datacallback = datacallback;

  while (tblget(&scc_procs, id = tblgetnexti(&scc_procs, id), &p))
    if ((p->control_fd < 0) && addrcmp(&scc_address, scc_addrlen, &p->address, p->addrlen)
	&& !settings.simulate)
      sccOpenControlConnection(p);
}

void sccCoprocessorSetup(int proc, scc_controlcallback_t controlcallback, scc_datacallback_t datacallback)
{
  scc_proc_t *p;

  lprintf(STREAM(sccl.notice), "Setting up coprocessor %d", proc);

  abortOnNULL(STREAM(sccl.err), tblget(&scc_procs, proc, &p));
  
  memcpy(&p->address, &scc_address, scc_addrlen);
  p->addrlen = scc_addrlen;
  p->controlcallback = controlcallback;
  p->datacallback = datacallback;
  
  if (scc_master.control_fd < 0 &&
      addrcmp(&scc_master.address, scc_master.addrlen, &p->address, p->addrlen) &&
      !settings.simulate)
    sccAwaitControlConnection();
}

void sccOpenControlConnection(scc_proc_t *p)
{
  scos_thread_t thread;
  struct sockaddr_in src, dst;
  int res, id=-1, opt=1;
  scc_proc_t *d;

  if (settings.spawn)
  {
    /* Spawn coprocessor process */
    char cmd[PATH_MAX], buf[PATH_MAX];
    unsigned char *bytes = (unsigned char *) &((struct sockaddr_in*)(&p->address))->sin_addr;
    getcwd(buf, PATH_MAX);

    sprintf(cmd, "ssh -f %d.%d.%d.%d 'cd %s && %s -n %s -m %s",
            bytes[0], bytes[1], bytes[2], bytes[3],
            buf, settings.argv[0], p->name, settings.machinesfile);

    if (settings.writetodisk)
      sprintf(&cmd[strlen(cmd)], " -w %d", p->id*1000+settings.writetodisk);
 
    sprintf(&cmd[strlen(cmd)], " > %s.stdout'", p->name);

    lprintf(STREAM(sccl.notice), "Spawning %s", cmd);

    system(cmd);
  }

  lprintf(STREAM(sccl.notice), "Initializing control connection to %s",
	  p->name);

  /* Initialise source address */
  memset(&src, 0, sizeof(src));
  src.sin_family = AF_INET;
  src.sin_addr.s_addr = INADDR_ANY;
  src.sin_port = 0;

  /* Initialise destination address */
  memset(&dst, 0, sizeof(dst));
  dst.sin_family = AF_INET;
  dst.sin_addr.s_addr = ((struct sockaddr_in*)&p->address)->sin_addr.s_addr;
  dst.sin_port = htons(SCC_MANAGEMENT_PORT);
  
  /* Open socket */
  abortOnNegative(STREAM(sccl.err), p->control_fd =
		  scos_socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
  abortOnNegative(STREAM(sccl.err),
		  scos_setsockopt(p->control_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)));
#ifdef NODELAY
  abortOnNegative(STREAM(sccl.err),
		  scos_setsockopt(p->control_fd, IPPROTO_TCP, TCP_NODELAY, &opt, sizeof(opt)));
#endif
  while (scos_bind(p->control_fd, (struct sockaddr*)&src, sizeof(src)) < 0)
  {
    fprintf(stderr, "%s couldn't bind to outgoing control socket", settings.name);
    fflush(stderr);
    sleep(1);
  }

  /* Setup transport */
  while (res = scos_connect(p->control_fd, (struct sockaddr*)&dst, sizeof(dst)),
	 (res < 0 && errno == ECONNREFUSED))
    usleep(10000);
  
  abortOnNegative(STREAM(sccl.err), res);

  /* Apply connection to other processors with the same address */
  id = -1;
  while (tblget(&scc_procs, id = tblgetnexti(&scc_procs, id), &d))
    if (d->control_fd < 0 &&
	!addrcmp(&d->address, d->addrlen, &p->address, p->addrlen))
      d->control_fd = p->control_fd;

  lprintf(STREAM(sccl.info), "Connected. Starting thread");

  /* Spawn result thread */
  scos_thread_create(&thread, sccIncomingControlThread, (void*)p);
}

void sccOpenDataConnection(scc_proc_t *p)
{
  struct sockaddr_in src, dst;
  int res, id=-1, opt=1;
  scc_proc_t *d;
  SOCKET fd;

  lprintf(STREAM(sccl.notice), "Initializing data connection to %s",
	  p->name);

  /* Initialise source address */
  memset(&src, 0, sizeof(src));
  src.sin_family = AF_INET;
  src.sin_addr.s_addr = INADDR_ANY;
  src.sin_port = 0;

  /* Initialise destination address */
  memset(&dst, 0, sizeof(dst));
  dst.sin_family = AF_INET;
  dst.sin_addr.s_addr = ((struct sockaddr_in*)&p->address)->sin_addr.s_addr;
  dst.sin_port = htons(SCC_DATA_PORT);
  
  /* Open socket */
  abortOnNegative(STREAM(sccl.err), fd =
		  scos_socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
  abortOnNegative(STREAM(sccl.err),
		  scos_setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)));
  opt = SCC_SOCKET_BUFSIZE;
  abortOnNegative(STREAM(sccl.err),
		  scos_setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &opt, sizeof(opt)));
  abortOnNegative(STREAM(sccl.err),
		  scos_setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &opt, sizeof(opt)));

  while (scos_bind(fd, (struct sockaddr*)&src, sizeof(src)) < 0)
  {
    fprintf(stderr, "%s couldn't bind to outgoing data socket", settings.name);
    fflush(stderr);
    sleep(1);
  }

  /* Setup transport */
  while (res = scos_connect(fd, (struct sockaddr*)&dst, sizeof(dst)),
	 (res < 0 && errno == ECONNREFUSED))
    usleep(10000);

  abortOnNegative(STREAM(sccl.err), res);
  p->outgoing_data_fd = fd;
  
  /* Apply connection to other processors with the same address */
  id = -1;
  while (tblget(&scc_procs, id = tblgetnexti(&scc_procs, id), &d))
    if (!addrcmp(&d->address, d->addrlen, &p->address, p->addrlen))
	d->outgoing_data_fd = p->outgoing_data_fd;
  
  lprintf(STREAM(sccl.info), "Connected");
}

void sccAwaitControlConnection(void)
{
  SOCKET fd, conn;
  struct sockaddr_in addr;
  int addrlen, opt=1;
  scos_thread_t thread;

  lprintf(STREAM(sccl.notice), "Awaiting control connection");

  /* Initialise address */
  memset(&addr, 0, sizeof(addr));
  addr.sin_family = AF_INET;
  addr.sin_addr.s_addr = INADDR_ANY;
  addr.sin_port = htons(SCC_MANAGEMENT_PORT);

  /* Open socket */
  abortOnNegative(STREAM(sccl.err), fd =
		  scos_socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
  abortOnNegative(STREAM(sccl.err),
		  scos_setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)));
#ifdef NODELAY
  abortOnNegative(STREAM(sccl.err),
		  scos_setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &opt, sizeof(opt)));
#endif
  while (scos_bind(fd, (struct sockaddr*)&addr, sizeof(addr)) < 0)
  {
    fprintf(stderr, "%s couldn't bind to incoming control socket", settings.name);
    fflush(stderr);
    sleep(1);
  }
  abortOnNegative(STREAM(sccl.err),
                  scos_listen(fd, 1));

  /* Wait for transport */
  addrlen = sizeof(addr);
  while (conn = scos_accept(fd, (struct sockaddr*)&addr, &addrlen),
	 (conn < 0 && errno == EINTR))
    usleep(10000);
  abortOnNegative(STREAM(sccl.err), conn);

  scos_close(fd);

  memcpy(&scc_master.address, &addr, addrlen);
  scc_master.addrlen = addrlen;
  scc_master.control_fd = conn;

  lprintf(STREAM(sccl.info), "Connected. Starting thread");

  /* Spawn result thread */
  scos_thread_create(&thread, sccIncomingControlThread, (void*)&scc_master);
}

void *sccIncomingConnectionThread(void *dummy)
{
#ifndef PSOS
  SOCKET fd, conn;
  struct sockaddr_in addr;
  int addrlen, opt=1;
  int id;
  scc_proc_t *p, *d;
  scos_thread_t thread;

  logRegisterThread("scc");
  lprintf(STREAM(sccl.info), "Started");

  /* Initialise address */
  memset(&addr, 0, sizeof(addr));
  addr.sin_family = AF_INET;
  addr.sin_addr.s_addr = INADDR_ANY;
  addr.sin_port = htons(SCC_DATA_PORT);

  /* Open socket */
  abortOnNegative(STREAM(sccl.err), fd =
		  scos_socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
  abortOnNegative(STREAM(sccl.err),
		  scos_setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)));
  opt = SCC_SOCKET_BUFSIZE;
  abortOnNegative(STREAM(sccl.err),
		  scos_setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &opt, sizeof(opt)));
  abortOnNegative(STREAM(sccl.err),
		  scos_setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &opt, sizeof(opt)));

  while (scos_bind(fd, (struct sockaddr*)&addr, sizeof(addr)) < 0)
  {
    fprintf(stderr, "%s couldn't bind to incoming data socket", settings.name);
    fflush(stderr);
    sleep(1);
  }
  abortOnNegative(STREAM(sccl.err), scos_listen(fd, 1));

  /* Wait for transport */
  while (1)
  {
    addrlen = sizeof(addr);
    while (conn = scos_accept(fd, (struct sockaddr*)&addr, &addrlen),
	   (conn < 0 && errno == EINTR))
      usleep(10000);
    abortOnNegative(STREAM(sccl.err), conn);

    /* Search for peer */
    id = -1;
    while (tblget(&scc_procs, id = tblgetnexti(&scc_procs, id), &p))
      if (!addrcmp((struct sockaddr*)&addr, addrlen, &p->address, p->addrlen))
        break;
  
    if (p)
    {
      p->incoming_data_fd = conn;
      lprintf(STREAM(sccl.info), "Connected. Starting thread");
      
      /* Apply connection to other processors with the same address */
      id = -1;
      while (tblget(&scc_procs, id = tblgetnexti(&scc_procs, id), &d))
	if (!addrcmp(&d->address, d->addrlen, &p->address, p->addrlen))
	  d->incoming_data_fd = p->incoming_data_fd;
      
      /* Spawn result thread */
      scos_thread_create(&thread, sccIncomingDataThread, (void*)p);
    }
    else
    {
      lprintf(STREAM(sccl.warning), "Unknown peer");
      scos_close(conn);
    }
  }  
#endif

  return NULL;
}

int sccGetPacket(SOCKET fd, char *buf)
{
  scc_hdr_t *hdr=(scc_hdr_t*)buf;
  int bytes, received;

  received = 0;
  do
  {
    bytes = scos_recv(fd, &buf[received], sizeof(scc_hdr_t)-received, MSG_WAITALL);
    received += bytes;
  } while (received < sizeof(scc_hdr_t) &&
	   ((bytes > 0) || (bytes < 0 && errno == EINTR)));

  returnOnError(STREAM(sccl.err), received != sizeof(scc_hdr_t));
  
  lprintf(STREAM(sccl.crawl), "Getting message type %d from %d for %d (%d bytes)",
	    hdr->type, hdr->source, hdr->destination, hdr->size);

  if (hdr->size)
  {
    received = 0;
    do
    {
      bytes = scos_recv(fd, &buf[sizeof(scc_hdr_t)+received], hdr->size-received,
			MSG_WAITALL);
      received += bytes;
    } while (received < hdr->size &&
	     ((bytes > 0) || (bytes < 0 && errno == EINTR)));

    returnOnError(STREAM(sccl.err), received != hdr->size);
  }

  return 0;
}

void *sccIncomingControlThread(void *proc)
{
  scc_proc_t *p = (scc_proc_t*)proc, *d;
  char buf[SCC_MAX_MESSAGE_SIZE];
  scc_hdr_t *hdr=(scc_hdr_t*)buf;

  sprintf(buf, "%s.ctrl", p->name);
  logRegisterThread(buf);

  lprintf(STREAM(sccl.info), "Started");

  while (!sccGetPacket(p->control_fd, buf))
  {
    if (hdr->destination == SCC_MASTER_PROCESSOR)
      d = &scc_master;
    else
      abortOnNULL(STREAM(sccl.err), tblget(&scc_procs, hdr->destination, &d));

    abortOnError(STREAM(sccl.err), !d->controlcallback);

    d->controlcallback(hdr);
  }
  
  lprintf(STREAM(sccl.warning), "Disconnected");
  exit(0);

  return NULL;
}

void *sccIncomingDataThread(void *proc)
{
  scc_proc_t *p = (scc_proc_t*)proc, *d;
  char buf[SCC_MAX_MESSAGE_SIZE];
  scc_hdr_t *hdr=(scc_hdr_t*)buf;

  sprintf(buf, "%s.data", p->name);
  logRegisterThread(buf);

  lprintf(STREAM(sccl.info), "Started");

  while (!sccGetPacket(p->incoming_data_fd, buf))
  {
    if (hdr->destination == SCC_MASTER_PROCESSOR)
      d = &scc_master;
    else
      abortOnNULL(STREAM(sccl.err), tblget(&scc_procs, hdr->destination, &d));

    abortOnError(STREAM(sccl.err), !d->datacallback);

    d->datacallback(hdr);
  }
  
  lprintf(STREAM(sccl.warning), "Disconnected");
  exit(0);

  return NULL;
}


int addrcmp(struct sockaddr *addr1, int addrlen1, struct sockaddr *addr2, int addrlen2)
{
  if (addrlen1 != addrlen2)
    return 1;

  if (addr1->sa_family != addr2->sa_family)
    return 1;

  switch (addr1->sa_family)
  {
    case AF_UNIX:
      return memcmp(addr1, addr2, addrlen1);
    case AF_INET:
      return ((struct sockaddr_in*)addr1)->sin_addr.s_addr != 
             ((struct sockaddr_in*)addr2)->sin_addr.s_addr;
    default:
      lprintf(STREAM(sccl.warning), "Unknown address family %d\n", addr1->sa_family);
      return 1;
  }
}

void sccWriteHeader(scc_hdr_t *hdr)
{
  printf("int source:%d, destination:%d;\n", hdr->source, hdr->destination);
  printf("unsigned short int type:%hu;\n", hdr->type);
  printf("unsigned lont int size:%lu;\n", hdr->size);
}

